#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__

#include "../mqcommon/logger.hpp"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/msg.pb.h"

#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <functional>

namespace mmq
{
    using ConsumerCallBack = std::function<void(const std::string, const BasicProperties *bp, const std::string)>;
    struct Consumer
    {
    public:
        using ptr = std::shared_ptr<Consumer>;
    public:
        Consumer()
        {
            DLOG("new Consumer: %p", this);
        }
        Consumer(const std::string &tag, const std::string &qname, bool ack_flag, const ConsumerCallBack &cb)
            : _tag(tag), _qname(qname), _auto_ack(ack_flag), _callback(cb)
        {
            DLOG("new Consumer: %p", this);
        }
        ~Consumer()
        {
            DLOG("del Consumer: %p", this);
        }
    public:
        std::string _tag;           // 消费者唯一标识
        std::string _qname;         // 消费者订阅的队列名称
        bool _auto_ack;             // 自动确认标志
        ConsumerCallBack _callback; // 回调函数
    };

    // 以队列为单元的消费者管理结构
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;
    public:
        QueueConsumer(const std::string &qname) : _qname(qname), _rr_seq(0)
        {}
        // 队列新增消费者
        Consumer::ptr Create(const std::string &tag, const std::string &qname, bool ack_flag, const ConsumerCallBack &cb)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断消费者是否重复
            for(auto &consumer : _consumers)
            {
                if(consumer->_tag == tag)
                {
                    return Consumer::ptr();
                }
            }
            // 没重复则新增----构造对象
            auto consumer = std::make_shared<Consumer>(tag, qname, ack_flag, cb);
            _consumers.push_back(consumer);
            return consumer;
        }
        // 队列移除消费者
        void Remove(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }
        // 队列获取消费者，RR轮转获取
        Consumer::ptr Choose()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if(_consumers.size() == 0) return Consumer::ptr();
            // 获取当前轮转下标
            int idx = _rr_seq % _consumers.size();
            _rr_seq++;
            // 获取对象并返回
            return _consumers[idx];
        }

        bool Empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }
        // 判断指定消费者是否存在
        bool Exists(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto &consumer : _consumers)
            {
                if(consumer->_tag == tag)
                {
                    return true;
                }
            }
            return false;
        }
        // 清理所有消费者    
        bool Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

        // 获取消费者数量
        size_t Count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        uint64_t _rr_seq;   // 轮转序号
        std::vector<Consumer::ptr> _consumers;
    };

    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
    public:
        ConsumerManager()
        {}

        void InitQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断是否重复
            auto it = _qconsumers.find(qname);
            if(it != _qconsumers.end()) return;
            // 新增
            auto qconsumer = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert({qname, qconsumer});
        }
        void DestroyQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.erase(qname);
        }
        Consumer::ptr Create(const std::string &tag, const std::string &qname, bool ack_flag, const ConsumerCallBack &cb)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("没有找到队列 %s 的消费者管理句柄", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->Create(tag, qname, ack_flag, cb); // 内部有自己的锁保护
        }
        void Remove(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("没有找到队列 %s 的消费者管理句柄", qname.c_str());
                }
                qcp = it->second;
            }
            qcp->Remove(tag);
        }
        Consumer::ptr Choose(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("没有找到队列 %s 的消费者管理句柄", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = it->second;
            }
            return qcp->Choose();
        }

        bool Empty(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("没有找到队列 %s 的消费者管理句柄", qname.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->Empty();
        }
        bool Exists(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("没有找到队列 %s 的消费者管理句柄", qname.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->Exists(tag);
        }  
        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }

        size_t ConsumerCount(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            QueueConsumer::ptr qcp;
            {
                auto it = _qconsumers.find(qname);
                if(it == _qconsumers.end())
                {
                    DLOG("找不到队列消费者的管理句柄，队列：%s 不存在", qname.c_str());
                    return 0;
                }
                qcp = it->second;
            }
            return qcp->Count();
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };

}

#endif